package com.scrcu.core.batch.service.impl;

import com.boarsoft.common.Util;
import com.boarsoft.common.util.InetUtil;
import com.scrcu.core.batch.constant.BatchConstants;
import com.scrcu.core.batch.entity.BatchEvent;
import com.scrcu.core.batch.entity.BatchExecution;
import com.scrcu.core.batch.entity.BatchInfo;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.Map;

public class BatchDispatcherBroadcastImpl extends BatchDispatcherImpl{
	
	private final static Logger log = LoggerFactory.getLogger(BatchDispatcherBroadcastImpl.class);

	@SuppressWarnings("unchecked")
	@Override
	public int start(String infoId, String entry, String params) {
		if (StringUtils.isEmpty(infoId)) {
			// return new ReplyInfo<String>(//
			// "Batch info id can not be empty");
			return 1;
		}
		BatchInfo info = infoBiz.get(infoId);
		if (info == null) {
			// return new ReplyInfo<String>(//
			// String.format("Batch info %s not found", infoId));
			return 2;
		}
		//
		// String key = info.toString();
		Date startTime = new Date();

		// 前次执行必须完成且成功
		BatchExecution exe = executionBiz.last(infoId);
		if (exe != null) {
			switch (exe.getStatus()) {
			case BatchConstants.EXECUTION_STATUS_NEW:
			case BatchConstants.EXECUTION_STATUS_RUNNING:
			case BatchConstants.EXECUTION_STATUS_SUSPEND:
			case BatchConstants.EXECUTION_STATUS_STOPPING:
				// return new ReplyInfo<String>(//
				// String.format("批量 %s 正在执行，不能重复调起", key));
				return 3;
			case BatchConstants.EXECUTION_STATUS_SUCCESS:
				break;
			case BatchConstants.EXECUTION_STATUS_FAILED:
			case BatchConstants.EXECUTION_STATUS_STOPPED:
				if (BatchConstants.INFO_FREEZE_BLOCK == info.getFreeze()) {
					// return new ReplyInfo<String>(//
					// String.format("批量 %s 的上一次执行未成功", key));
					return 4;
				}
				break;
			}
			// 检查批量的冻结时间，为0表示不冻结
			int freeze = info.getFreeze();
			if (freeze > 0) {
				// 上一次开始执行时间
				Date st = Util.str2date(exe.getStartTime(), Util.STDDTMF);
				// 计算两个时间的总时毫秒差，并转换为分钟）
				long m = (startTime.getTime() - st.getTime()) / (1000 * 60);
				if (m < freeze) {
					// return new ReplyInfo<String>(String.format(//
					// "批量 %s 需要再过 %d 分钟，才可以被再次调起", key, freeze - m));
					return 5;
				}
			}
		}
		
		// 检查对应batchstep中的终态个数与广播个数是否相符
		/*
		if (exe != null) {
			exe.getInfoId();
			if (exe.getAddr().indexOf("broadcast") > -1) {
				int broadcastNum = Integer.valueOf(exe.getAddr().split(":")[1]);
				int endNum = eventBiz.listEnd(exe.getId()).size();
				if (broadcastNum > endNum) {
					log.info("Last batch execution {} is not end。 Lack of end batch event");
					return 3;
				}
			}
		}
		*/

		// 没有传入参数就使用默认参数
		if (Util.strIsEmpty(params)) {
			params = info.getParams();
		}
		// 登记批量执行记录
		exe = new BatchExecution();
		exe.setAddr(InetUtil.getAddr());
		exe.setCatalog(info.getCatalog());
		exe.setCode(info.getCode());
		exe.setEntry(entry);
		exe.setInfoId(infoId);
		// exe.setNo(no);
		exe.setParams(params);
		exe.setStartTime(startTime);
//		exe.setEndTime("");
		exe.setStatus(BatchConstants.EXECUTION_STATUS_NEW);
		exe.setType(BatchConstants.EXECUTION_TYPE_START);
//		exe.setError("");
//		exe.setStep("");
		
		String exeId = executionBiz.save(exe);

		// 开始执行
		log.info("Start batch execution {}", exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("start");
		evt.setExeId(exe.getId());
		// evt.setParams(params);
		evt.setStatus(BatchConstants.EXECUTION_STATUS_NEW);
		// evt.setStep(step);
		eventBiz.save(evt);

		try {
			// 广播，服务地址0到n个
			Map<String,Object> map = (Map<String,Object>)broker.startAll(exe);
			log.info("Batch execution {} be post to {}", exe, map.keySet());
			// 判断是否成功调起rpc服务.若没有成功调起的应答，则批量执行状态为失败。
			exe.setStatus(BatchConstants.EXECUTION_STATUS_FAILED);// 执行失败
			int broadcastNum = map.keySet().size(); //广播收到应答数
			if (broadcastNum > 0) {
				for (String key : map.keySet()) {
					if (!(map.get(key) instanceof Throwable)) {
						exe.setStatus(BatchConstants.EXECUTION_STATUS_SUCCESS);// 广播成功
						continue;
					}
					broadcastNum--;// 未广播成功，计数减1
				}
			}
			exe.setAddr("broadcast:"+broadcastNum);
			executionBiz.save(exe);
			// return new ReplyInfo<String>();
			return 0;
		} catch (Throwable e) {
			log.error("Error on post batch execution {}", exeId, e);
			exe.setStatus(BatchConstants.EXECUTION_STATUS_FAILED);
			exe.setError(e.getMessage());
			executionBiz.save(exe);
			// return new ReplyInfo<String>(e.getMessage());
			return 6;
		}
	}
	
	@Override
	public int resume(String exeId) {
		// 不允许执行
		return BatchConstants.EXECUTION_ERROR_UNKNOWN;
	}
	
	@Override
	public int cancel(String exeId, boolean force) {
		// 不允许执行
		return BatchConstants.EXECUTION_ERROR_UNKNOWN;
	}
	
	@Override
	public int redo(String exeId) {
		// 不允许执行
		return BatchConstants.EXECUTION_ERROR_UNKNOWN;
	}
	
	@Override
	public void onBegin(String exeId, String addr) {
		// 记录批量超时事件（由批量应用端主动通知）
		BatchEvent evt = new BatchEvent();
		evt.setAddr(addr);
		evt.setContent("begin");
		evt.setExeId(exeId);
		evt.setStatus(BatchConstants.EXECUTION_STATUS_RUNNING);
		eventBiz.save(evt);
	}
	
	@Override
	public void onStep(String exeId, String step, String addr) {
		log.info("Batch execution {} step {} at {}", exeId, step, addr);
		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStatus(BatchConstants.EXECUTION_STATUS_RUNNING);
		exe.setStep(step);
		exe.setAddr(addr);
		// executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("step");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStatus(exe.getStatus());
		evt.setStep(step);
		eventBiz.save(evt);
	}
	
	@Override
	public void onFailed(String exeId, String step, String addr, String error, String params) {
		log.info("Batch execution {}/{} failed at {}", exeId, step, addr);
		
		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStep(step);
		exe.setAddr(addr);
		exe.setStatus(BatchConstants.EXECUTION_STATUS_FAILED);
		exe.setEndTime(new Date());
		exe.setError(error);
		//executionBiz.save(exe);
		
		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("failed");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStep(step);
		evt.setStatus(exe.getStatus());
		eventBiz.save(evt);
		
		// 更新任务状态
		executionBiz.updateStatusFailed(exeId);
	}
	
	@Override
	public void onSuspend(String exeId, String step, String addr) {
		log.info("Batch execution {} suspend {} at {}", exeId, step, addr);

		// 更新任务状态
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStatus(BatchConstants.EXECUTION_STATUS_SUSPEND);
		exe.setStep(step);
		exe.setAddr(addr);
		// executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("suspend");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStatus(exe.getStatus());
		// evt.setStep(step);
		eventBiz.save(evt);
	}
	
	@Override
	public void onSuccess(String exeId, String addr) {
		log.info("Batch execution {} completed at {}", exeId, addr);
		
		BatchExecution exe = executionBiz.get(exeId);
		exe.setStatus(BatchConstants.EXECUTION_STATUS_SUCCESS);
		exe.setStep(null);
		exe.setAddr(addr);
		exe.setEndTime(new Date());
//		executionBiz.save(exe);

		// 记录事件
		BatchEvent evt = new BatchEvent();
		evt.setAddr(exe.getAddr());
		evt.setContent("success");
		evt.setExeId(exeId);
		// evt.setParams(params);
		evt.setStatus(exe.getStatus());
		// evt.setStep(step);
		eventBiz.save(evt);
		
		// 更新任务状态
		executionBiz.updateStatusFailed(exeId);
		executionBiz.updateStatusSuccess(exeId);
	}
}
